package cn.bonoon.kernel.io;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.bonoon.kernel.io.synchronize.NeedExitTaskException;

/**
 * 多线程的工作任务表
 * 这个父类适合任务队列的处理，即一个任务处理完成就下一个任务，并且这些任务使用多线程进行处理
 * 
 * @author jackson
 *
 * @param <T>
 */
public abstract class AbstractTaskThread<T> implements Runnable{

	protected final Logger log = LoggerFactory.getLogger(this.getClass());

	/** 如果线程里出现异常，则最大的尝试的次数 */
	private final static int THREAD_ERROR_MAX_TRY_COUNT = 3;
	private final static int DEFAULT_THREAD_MAX_SIZE = 5;
	
	private final static int DEFAULT_LARGE_TASK_SIZE = 500;
	
	/** 当前同步任务使用的最大的线程，允许动态调整线程；但最高的线程不要超过这个值的限制 */
	private final int threadMaxSize;
	
	/** 当前正在执行的线程数组 */
	private final Thread[] threads;
	
	/** 用于线程之间的同步 */
	private final Object LOCK = new Object();

	/** 当前活动的线程 */
	private int activityThread;
	
	/** 任务的执行时间，如果为0，则表示不限制时间 */
	private long taskExecuteTimes;
	
	/** 一次启动的话，要初始化多少个线程 */
	private int onceStartTaskCount;
	
	/** 第一次初始化的时候，启动多少个线程 */
	private int initStartTaskCount;

	/** 当前任务是否结束 */
	protected boolean running;
	
	protected final List<T> needSynchronizeItems = new ArrayList<>(200);

	/** 任务的主线程是否等待线程的唤醒 */
	private final boolean mainThreadWait4Notify;
	
	/** 当任务完成时，即没有待处理的内容项的时候，子任务线程是否退出 */
	private final boolean quitThreadIfTasksFinish;

	/** 工作任务线程的优先级;
	 * 每个类都有自己的优先级，一般property用1-10的整数表示，默认优先级是5，优先级最高是10；
	 * 优先级高的线程并不一定比优先级低的线程执行的机会高，只是执行的机率高；
	 * 默认一个线程的优先级和创建他的线程优先级相同； */
	private int priority;
	
	/** 表示缓存的大小，如果没有指定，则表示来一个数据就处理一个数据；
	 * 如果指定为大于0的数值，则表示：当"所有"线程处于等待状态，需要被处理的数据的数量
	 * 必须达到指定的最小缓存的大小，才会被启动处理 */
	private int batchTaskSize = 0;
	
	/** 需要更多的线程来处理任务 */
	private int largeTaskSize = DEFAULT_LARGE_TASK_SIZE;
	
	/** 处理的总的任务数 */
	private long taskItemTotal;
	/** 处理成功的任务数 */
	private long taskItemSuccess;
	/** 最小的处理时间 */
	private long taskTimeMin = Long.MAX_VALUE;
	/** 任务最长的处理时间 */
	private long taskTimeMax;
	/** 任务平均处理时间 */
	private long taskTimeAvg;
	/** 任务执行的总时间，只有在任务执行的时候才计时 */
	private long taskTimeTotal;
	
	public String currentState(){
		if(!running) return "当前任务已经停止！";
		if(taskItemTotal == 0) return "没有执行任务！";
		return "任务数(成功/总数)：" + taskItemSuccess + "/" + taskItemTotal
				+ ";任务耗时(总耗时/最大/平均/最小)：" + taskTimeTotal + "/" + taskTimeMax + "/" + taskTimeAvg + "/" + taskTimeMin
				+ ";工作线程/等待线程：" + activityThread + "/" + waitingThread + ";";
	}
	
	/**
	 * 构造函数，必须要创建对象时指定的属性
	 * 
	 * @param mainThreadWait4Notify 参考：{@link #mainThreadWait4Notify}
	 * @param quitThreadIfTasksFinish 参考：{@link #quitThreadIfTasksFinish}
	 */
	protected AbstractTaskThread(boolean mainThreadWait4Notify, boolean quitThreadIfTasksFinish){
		this(mainThreadWait4Notify, quitThreadIfTasksFinish, DEFAULT_THREAD_MAX_SIZE);
	}
	
	protected AbstractTaskThread(boolean mainThreadWait4Notify, boolean quitThreadIfTasksFinish, int threadMaxSize){
		this.mainThreadWait4Notify = mainThreadWait4Notify;
		this.quitThreadIfTasksFinish = quitThreadIfTasksFinish;
		this.threadMaxSize = threadMaxSize;
		this.threads = new Thread[threadMaxSize];
	}
	
	protected boolean starting(){ return true; }
	
	protected void started(){}
	
	protected boolean stoping(){ return true; }
	
	/** 停止工作任务时会被调到函数，理论上必须被调用 */
	protected void stoped(){
		needSynchronizeItems.clear();
		Arrays.fill(threads, null);
	}

	/** 添加待处理的列表 */
	public void add(T[] items){
		synchronized (LOCK) {
			for(T item : items){
				if(checked(item)){
					needSynchronizeItems.add(item);
				}
			}
			
			if(!needSynchronizeItems.isEmpty()){
				lockNotify();
			}
		}
	}
	
	/** 已经使用了同步锁 */
	public void add(T item){
		if(checked(item)){
			synchronized (LOCK) {
				needSynchronizeItems.add(item);
				lockNotify();
			}
		}
	}
	
	private void lockNotify(){
		if(activityThread > waitingThread || batchTaskSize < needSynchronizeItems.size()){
			/* 
			 * "activityThread > waitingThread"表示有线程正在处理任务
			 * "minBatchSize < needSynchronizeItems.size()"如果是这种情况，则表示当前所有的任务线程正在等待
			 * 或者工作任务已经停止了，有效的情况是：quitThreadIfTasksFinish=false 的情况
			 * 
			 * 只有在有数据的情况下
			 * 通知所有正在等待的线程，可以开始抢了
			 * 注意添加单个和添加多个线程锁定的不同 
			 */
			LOCK.notifyAll();
		}
	}

	/** 用于判断将要加入的工作任务信息是否正确，工作线程只处理检查通过的任务信息 */
	protected boolean checked(T item){ return null != item; }

	/** 尝试停止任务，完成停止，所有子线程都需要退出 */
	private void tryStop(boolean waiting4taskStop) throws Exception{
		
		synchronized (LOCK) {
			/*
			 * 1. 表示工作本身已经停止了，不需要再进行停止的操作
			 * 2. 表示子类的停止检查结果是不验证停止当前的工作任务线程
			 */
			if(!running || !stoping()) return;
				
			/*
			 * 设置为停止任务状态
			 * 为了防止两个线程同时调用停止操作，所以在这里设置为同步执行，如果检查到已经停止了，则直接退出
			 */
			running = false;
		
			/*
			 * 这里需要唤醒所有的子任务正在等待的线程
			 * 如果不唤醒，则有些子任务线程会一直处于等待状态没有被释放
			 */
			LOCK.notifyAll();
			
		}
		
		if(waiting4taskStop){
			synchronized(this){
				/*
				 * 这个主线程要等待所有子线程的任务正常的处理完成再退出
				 * 一个子线程执行的时间可能会比较长，所以主线程需要等待子线程的结束；
				 * 可能线程结束后还需要处理一些后继的工作
				 * 
				 * 如：正在用FTP方式传输文件到服务器，如果这个时间得到停止任务的通知，
				 * 则在请求停止任务的线程上，需要等待文件正常的传输完毕，再做收尾的处理工作
				 * 
				 */
				wait(taskExecuteTimes);
			}
		}else{
			Thread.sleep(500);
		}
		//线程结束的收尾处理工作
		stoped();
		
	}

	public final void stop(){
		stop(true);
	}
	
	/** 停止任务，在停止任务的时候，需要指定请求停止任务的线程是否需要等待所有的线程完成后再退出 */
	public final void stop(boolean waiting4taskStop){
		try{
			if(mainThreadWait4Notify){
				synchronized(this){
					notifyAll();
				}
			}else{
				tryStop(waiting4taskStop);
			}
			
		}catch (Exception e) {
			log.error("停止任务时出现异常！", e);
		}
	}
	
	/** 是否需要更多的处理线程 */
	private void moreTaskThread(int count){
		if(activityThread >= threadMaxSize) return;
		int currentCount = 0;
		for(int i = 0; i < threadMaxSize; i++){
			Thread thread = threads[i];
			if(null == thread || !thread.isAlive()){
				threads[i] = thread = new Thread(this);
				thread.setName("file synchronize thread:" + activityThread++);
				if(0 < priority && priority <= 10){
					//如果有定义了线程的优先级，则设置启动时线程的优先级
					thread.setPriority(priority);
				}
				currentCount++;
				//成功启动或重新启动了一个线程
				if(count > 0 && currentCount >= count){
					break;
				}
			}
		}
		
//		//启动线程
//		for(Thread thread : threads){
//			thread.setName("file synchronize thread:" + activityThread++);
//			if(0 < priority && priority <= 10){
//				//如果有定义了线程的优先级，则设置启动时线程的优先级
//				thread.setPriority(priority);
//			}
//			
//			thread.start();
//		}
	}
	
	public final void start(){
		try{
			synchronized (LOCK) {
				
				if(running || !starting()){
					//当前的处理线程正在进行，不需要再重新启动
					return;
				}
			
				//准备线程，0表示启动最大线程
				moreTaskThread(initStartTaskCount);
			
				running = true;
			}
		
			
			/*
			 * 到这里，线程已经启动完成，子类可以处理线程完成后的一些事情；
			 * 子线程启动完成后，任务的主线程就要进入休眠状态，直接规定的时间或
			 * 所有子线程的任务已经完成，或者由外部线程进行了中断操作
			 */
			started();
			
			if(mainThreadWait4Notify){
				synchronized(this){
					//线程启动完成后，就一直在这里等待
					wait(taskExecuteTimes);
				}
					
				/*
				 * 所有任务线程都完成了，或规定的任务执行时间完成；
				 * 这里都会尝试停止所有的工作线程
				 */
				tryStop(true);
			}
		}catch (Exception e) {
			log.error("执行任务时出错！", e);
		}
	}
	
	private int waitingThread;
	
	@Override
	public void run(){

		int errorCount = 0;
		
		//如果已经stop，则退出整个线程
		TASK : while(running){
			try{
				T currentTask = null;
				synchronized (LOCK) {
					if(needSynchronizeItems.isEmpty()){
						waitingThread++;
						try{
							if(quitThreadIfTasksFinish && waitingThread >= activityThread){
								//所有的线程都处于等待状态，则表示这些目录的处理已经完成，可以退出了
								break TASK;
							}
							
							//如果有更多的线程处于活动中，则本线程在这里等待别的线程处理结束
							LOCK.wait();
						}catch (Exception e) {
							/*
							 * 如果是在线程等待时出现的异常，则必须把
							 * waitingThread--;
							 * 给还回去
							 */
							throw e;
						}finally{
							waitingThread--;
						}
						continue TASK;
					}
					
					/*
					 * 1. 成功取得一个需要处理的任务的信息
					 * 2. 如果这里还有很多工作要处理，则尝试启动更多的任务线程来处理
					 */
					currentTask = needSynchronizeItems.remove(0);
					
					if(onceStartTaskCount > 0 && needSynchronizeItems.size() > largeTaskSize){
						moreTaskThread(onceStartTaskCount);
					}
				}
				
				taskItemTotal++;
				long start = System.currentTimeMillis();
					
				try{
					if(executeTask(currentTask)){
						taskItemSuccess++;
					}
					
					/*
					 * 成功的完成了一个任务
					 * 通知jvm释放内存，也表示这个不需要再被处理
					 * 重置异常统计次数
					 */
					currentTask = null;
					errorCount = 0;
					
				}catch(NeedExitTaskException exitTask){
					/*
					 * 如：文件同步服务器不可用的话，则要退出处理
					 * 一般是由网络问题造成的
					 */
					log.error("执行任务出现异常，任务无法继续执行！", exitTask);
					
					//需要直接退出线程
					break TASK;
				}catch (Exception e) {
					// 如果是文件同步服务器出现异常，则无法把文件上传到文件同步服务器上，这个时间应该如何处理
					throw e;
				}finally{
					if(null != currentTask){
						/*
						 * 处理不成功时，把当前正在处理的文件夹还回去，让他能被下一个线程处理
						 * 还需要通知下其它线程进行处理，如果其它线程有在等待的
						 */
						add(currentTask);
					}

					long currentTime = System.currentTimeMillis() - start;
					taskTimeMax = Math.max(taskTimeMax, currentTime);
					taskTimeMin = Math.min(taskTimeMin, currentTime);
					taskTimeTotal += currentTime;
					taskTimeAvg = taskTimeTotal / taskItemTotal;
					
				}
			}catch (Exception e) {
				log.error("同步文件时出现异常！", e);
				
				try {
					Thread.sleep(500);
				} catch (InterruptedException ie) {
					//线程本身都出错了。
					log.error("线程异常，无法执行线程的休眠！", ie);
				}
				
				errorCount++;
				if(errorCount > THREAD_ERROR_MAX_TRY_COUNT){
					break TASK;
				}
			}
		}

		try {
			//有一个线程退出
			activityThread--;
			
			synchronized (LOCK) {
				
				/*
				 * 只有当前线程处理活动状态，其它线程已经退出或处于wait状态的，
				 * 这个时候进入这里，则表示已经没有需要处理的任务了，可以退出本
				 * 次的工作任务
				 */
				LOCK.notifyAll();
			}
			if(activityThread <= 0){
				synchronized (this) {
					/*
					 * 所有的子线程任务已经退出
					 * 通知主线程退出
					 */
					notifyAll();
				}
				//保证不要让活动线程出现负数的情况
				activityThread = 0;
			}
			
			//处理一些收尾的工作
			closeTask();
		
		} catch (Exception e) {
			log.error("关闭任务出现异常！", e);
		}
			
		log.info("当前线程[" + Thread.currentThread().getName() + "]退出！");
	}
	
	/** 正在被执行的任务 */
	protected abstract boolean executeTask(T task) throws Exception;

	/** 关闭一个线程的文件同步任务 */
	protected void closeTask() throws Exception { }

//	/** 参考：{@link #threadMaxSize} */
//	public void setThreadMaxSize(int threadMaxSize) {
//		this.threadMaxSize = threadMaxSize;
//	}
	
	/** 参考：{@link #priority} */
	public void setPriority(int priority) {
		this.priority = priority;
	}
	public void setBatchTaskSize(int batchTaskSize) {
		this.batchTaskSize = batchTaskSize;
	}
}
